package com.ruoyi.mstore.consumer;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import com.ruoyi.mstore.domain.RescueMessage;
import com.ruoyi.mstore.service.RescueTaskService;
import lombok.Builder;
import lombok.Data;
import lombok.RequiredArgsConstructor;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.data.geo.Circle;
import org.springframework.data.geo.Distance;
import org.springframework.data.geo.Metrics;
import org.springframework.data.geo.Point;
import org.springframework.data.redis.connection.RedisGeoCommands;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.HashMap;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Component
@RequiredArgsConstructor
public class MerchantConsumer {
    private final StringRedisTemplate redisTemplate;
    private final SimpMessagingTemplate messagingTemplate;
    private final ObjectMapper objectMapper;
    private final RescueTaskService rescueTaskService;

    @RabbitListener(queues = {"rescue.queue.1km", "rescue.queue.2km", "rescue.queue.3km"})
    @Transactional
    public void handleRescueMessage(String message, Channel channel,
                                    @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        try {
            RescueMessage rescueMsg = objectMapper.readValue(message, RescueMessage.class);

            // 广播给对应区域的商家
            notifyMerchants(rescueMsg);

            // 等待商家响应（模拟实际抢单流程）
            boolean accepted = waitForAcceptance(rescueMsg);

            if (accepted) {
                // 处理接单逻辑
                processAcceptance(rescueMsg);
                channel.basicAck(tag, false);
            } else {
                // 拒绝消息（进入延迟队列）
                channel.basicReject(tag, false);
            }
        } catch (Exception e) {
            System.out.println("处理消息时发生错误: " + e.getMessage());
        }
    }

    private void notifyMerchants(RescueMessage message) {
        // 根据地理围栏获取目标商家ID列表
        List<String> merchantIds = findNearbyMerchants(message.getLocation());

        merchantIds.forEach(merchantId -> {
            messagingTemplate.convertAndSendToUser(
                    merchantId,
                    "/queue/rescue-requests",
                    buildNotification(message)
            );
        });
    }

    private RescueNotification buildNotification(RescueMessage message) {
        return RescueNotification.builder()
                .messageId(message.getMessageId())
                .rescueType(message.getRescueType())
                .location(message.getLocation())
                .vehicleInfo(message.getVehicleInfo())
                .timestamp(System.currentTimeMillis())
                .expireIn(15) // 15秒内响应
                .build();
    }

    private boolean waitForAcceptance(RescueMessage message) {
        String lockKey = "rescue:accept:" + message.getMessageId();
        try {
            // 等待商家响应（实际应为异步事件驱动）
            return Boolean.TRUE.equals(redisTemplate.opsForValue()
                    .setIfAbsent(lockKey, "processing", 15, TimeUnit.SECONDS));
        } finally {
            redisTemplate.delete(lockKey);
        }
    }

    private void processAcceptance(RescueMessage message) {
        // 更新数据库状态
        rescueTaskService.acceptRescueTask(message.getMessageId(), message.getMemberId().toString());

        // 通知用户
        notifyUserAcceptance(message.getMemberId().toString(), message.getMessageId());
    }

    private void notifyUserAcceptance(String memberId, String messageId) {
        messagingTemplate.convertAndSendToUser(
                memberId,
                "/queue/rescue-status",
                new HashMap<String, Object>() {{
                    put("status", "ACCEPTED");
                    put("messageId", messageId);
                    put("timestamp", System.currentTimeMillis());
                }}
        );
    }


    // 实现地理围栏查询
    private List<String> findNearbyMerchants(RescueMessage.GeoLocation location) {
        Point point = new Point(location.getLongitude(), location.getLatitude());
        Distance distance = new Distance(2, Metrics.KILOMETERS); // 2公里范围

        return redisTemplate.opsForGeo()
                .radius("merchant:locations",
                        new Circle(point, distance),
                        RedisGeoCommands.GeoRadiusCommandArgs.newGeoRadiusArgs().includeCoordinates())
                .getContent()
                .stream()
                .map(geoResult -> geoResult.getContent().getName().toString())
                .collect(Collectors.toList());
    }



    @Data
    @Builder
    private static class RescueNotification {
        private String messageId;
        private Integer rescueType;
        private RescueMessage.GeoLocation location;
        private RescueMessage.VehicleInfo vehicleInfo;
        private Long timestamp;
        private Integer expireIn;
    }
}